﻿/*************************************************************************
** Copyright(c) 2016-2025  xktesla
** All rights reserved.
** Name		: Msger.h（之前的Mailer）
** Desc		: 一个高级订阅器，接口简单但强大。
**              Msg_XX_Id：邮件id，需要使用DEF_MSG_XXXX来定义
**              Msger：订阅器主类
**              IExecutor：发布器基类
**              IReviewer：接收器基类
** Author	: xktesla@2020-12-19 09:19:30
*************************************************************************/

#ifndef _1599C098_414D_4A2A_B59B_EF39E0151B90
#define _1599C098_414D_4A2A_B59B_EF39E0151B90

#include "depends/incfile.h"
//#include "AnyT.h"
#include "xCores.h"
#include "xUtils.h"
#include "ISingleton.h"

namespace x2lib
{
    const uint16_t Msg_Min_Id = 0x1000;
    const uint16_t Msg_Max_Id = 0xEFFF;
    const uint16_t Msg_Base_Id = 0x1001;
    const uint16_t Msg_None_Id = 0x0000;

    /*************************************************************************
    ** Desc     : 订阅器。支持多个订阅者和多个发布者
    ** Author   : xktesla@2020-12-19
    *************************************************************************/
    class Msger:
        public ISingleton<Msger, bool>,
        public xCores::Engine
    {
    public:
        friend class ISingleton<Msger, bool>;

        /*************************************************************************
        ** Desc     : 邮件【数据包】
        ** Author   : xktesla@2020-12-19
        *************************************************************************/
        struct MsgX
        {
            friend class Msger;
            friend class MsgPool;

        public:
            virtual MsgX* Reset() { return this; };
            virtual MsgX* Clone() const { return nullptr; };

            const uint32_t uDbgId; // 方便调试
            const uint16_t uMsgId;
            const bool isPost;
            // 废弃isTell，是否Feed由Reciever是否监听决定
            // bool isTell; // 是否需要回调通知，供外部使用，仅Post有效
            int iError; // 执行结果错误码，建议0成功，-1未知错误，1xx表示ReqXXX错误，2xx表示EvlXXX错误，3xx表示RepXXX错误

        protected:
            MsgX(uint16_t uMsgId, bool isPost, bool auTell/*, AnyT& atArgs = *static_cast<AnyT*>(nullptr)*/);

            const bool auTell; // Post的消息被所有IExecutor处理完后，是否自动反馈给IReviewer
            const bool isTake;
            static uint32_t __gen_DbgId__();
        };

        /*************************************************************************
        ** Desc     : 发布者，从订阅器线程回调
        ** Author   : xktesla@2020-12-19
        *************************************************************************/
        class IExecutor
        {
        public:
            // 返回值表示是否终止消息流转，同时也将无法触发Tell反馈，禁止在回调函数中调用MsgBind/Send
            virtual bool OnMsgExec(const MsgX* pMsg) = 0;
            //virtual void OnSendMail(const MsgX* pMsg) = 0;

            // 是否UnTake
            //virtual bool OnPostMail(const MsgX* pMsg) = 0;

            // 可变参最后一个值必须为NONE
            //1. addMode为true: 添加传入的Topic
            //2. addMode为false: 删除传入的Msg_XX_Id，若第一个可变参为Msg_None_Id则删除所有
            bool MsgBind(IExecutor* pExecutor, bool addMode, ...);
        };

        /*************************************************************************
        ** Desc     : 订阅者，从订阅器线程回调，并接收所有Post反馈消息和Tell消息
        ** Author   : xktesla@2020-12-19
        *************************************************************************/
        class IReviewer
        {
        public:
            // 先绑定先接收，禁止在回调函数中调用MsgBind/Send
            virtual bool/*是否终端流转，同时会导致无法UnTake*/ OnMsgTell(const MsgX* pMsg) = 0; // OnMsgTell内需要同步执行，因为之后可能会UnTake

            // 可变参最后一个值必须为NONE
            //1. addMode为true: 添加传入的Topic
            //2. addMode为false: 删除传入的Msg_XX_Id，若第一个可变参为Msg_None_Id则删除所有
            bool MsgBind(IReviewer* pReviewer, bool addMode, ...);
        };

        /*************************************************************************
        ** Desc     : 异步请求，向发布者请求一个msg任务。并接受（通过IReviewer）执行反馈
        ** Param    : [in] uMsgId
        ** Param    : [in] arArgs
        ** Param    : [in] isTell 是否需要回调通知
        ** Return   :
        ** Author   : xktesla@2020-12-19
        *************************************************************************/
        //void Post(uint16_t uMsgId, const AnyT& arArgs = *static_cast<AnyT*>(nullptr), bool isTell = true)
        void Post(MsgX* pMsg);

        /*************************************************************************
        ** Desc     : 同步请求，向发布者请求一个msg任务
        ** Param    : [in] pMsg Send不能使用MsgPool里的msg
        ** Return   : 返回传入的指针
        ** Author   : xktesla@2020-12-19
        *************************************************************************/
        //const AnyT* Send(uint16_t uMsgId, AnyT& atArgs = *static_cast<AnyT*>(nullptr))
        const MsgX* Send(MsgX* pMsg, bool isDirect = false);

        /*************************************************************************
        ** Desc     : 异步通知，向订阅者(IReviewer)通知一个msg结果
        ** Param    : [in] pMsg 若pMsg来自于MsgPool则在所有IReviewer处理完成后会自动UnTake
        ** Return   :
        ** Author   : xktesla@2020-12-19
        *************************************************************************/
        void Tell(MsgX* pMsg, bool isDirect = false);

    protected:
        /**
        * @brief        初始化
        * @param[in]    iTwoThread true将IExecutor/IReviewer运作在不同线程，否则运作在同一线程
        * @author       xktesla
        */
        Msger(bool inTwoThread);
        virtual ~Msger();
        virtual bool initInst() override;
        virtual void unInitInst() override;

        bool/*空轮询*/ notifyExecutor();

        bool/*空轮询*/ notifyReviewer();

        // 由Ignite触发
        virtual void onEngineExec(uint32_t id, void* vData) override;
        // 由Stall触发
        virtual void onEngineExit(uint32_t id, void* vData) override;


        bool m_inTwoThread;
        int ID_CYLID0, ID_CYLID1, ID_CYLID2;
        std::map<uint16_t, std::unordered_set<IExecutor*>> m_mspExecutor; // 发布者集合
        std::map<uint16_t, std::unordered_set<IReviewer*>> m_mspReviewer; // 订阅者集合
        std::queue<MsgX*> m_quepMsging; // 待处理队列
        std::queue<MsgX*> m_quepMsgend; // 处理完队列
        xCores::Mutex m_mtxMappExecutor;
        xCores::Mutex m_mtxMappReviewer;
        xCores::Mutex m_mtxQuepMsging;
        xCores::Mutex m_mtxQuepMsgend;
        xCores::Signal m_SigMsging;
        //xCores::Signal m_SigMsgend;
        bool m_isRunning;
        int m_nExit;
    };

    class MsgPool:
        public ISingleton<MsgPool>
    {
        friend class ISingleton<MsgPool>;
        friend class Msger;

    public:
        template<class T> T* Take()
        {
            m_mtxPool.Lock();
            for (auto& it : m_setMsg)
            {
                if (dynamic_cast<T*>(it) && !it->isTake)
                {
                    const_cast<bool&>(it->isTake) = true;
                    it->Reset(); // 防止调用者在UnTake后非法操作
                    m_mtxPool.Unlock();
                    return (T*)it;
                }
            }

            Msger::MsgX* msg = new T;
            const_cast<bool&>(msg->isTake) = true;
            m_setMsg.insert(msg);
            m_mtxPool.Unlock();
            return (T*)msg;
        }

        // 一般只适合Send消息调用，但不禁止Post消息调用，因为可能用户在Post和Tell都设置了autoUnTake=false
        void UnTake(Msger::MsgX* msg);

    protected:
        std::set<Msger::MsgX*> m_setMsg;
        //std::vector<std::pair<Msger::MsgX*, int>> m_vkUnTake; // 针对Post类型msg设计
        xCores::Mutex m_mtxPool;
    };
}

#define DEF_MSG_BEGIN		namespace x2lib {


#define DEF_MSG_ITEM_BEGIN(id, byPost, autoTell, tpcName)				 							                    \
    constexpr uint16_t Msg_##tpcName##_Id = (Msg_Base_Id + id);                                                         \
    struct Msg_##tpcName: public Msger::MsgX            											                    \
    {																								                    \
        Msg_##tpcName() :Msger::MsgX((Msg_Base_Id + id), byPost, autoTell)           				            		\
        {                                                                                                               \
            Reset();																									\
        }                                                                                                               \
        virtual MsgX* Clone() const override																		    \
        {                                                                                                               \
            return new Msg_##tpcName(*this);                                                                            \
        };
        //argList;
#define DEF_MSG_ITEM_RESET(tpcName)                                                                                     \
        virtual Msg_##tpcName* Reset() override																		    \
        {                                                                                                               \
            this->iError = 0;
            //argInit;
#define DEF_MSG_ITEM_END(tpcName)                                                                                       \
            return this;                                                                                                \
        };						                                                                                        \
    };


#define DEF_MSG_ITEM_BASE_BEGIN(id, byPost, autoTell, tpcName, struBase)				 							    \
    constexpr uint16_t Msg_##tpcName##_Id = (Msg_Base_Id + id);                                                         \
    struct Msg_##tpcName: public Msger::MsgX, public struBase            											    \
    {																								                    \
        Msg_##tpcName() :Msger::MsgX((Msg_Base_Id + id), byPost, autoTell)           				            		\
        {                                                                                                               \
            Reset();																									\
        }                                                                                                               \
        virtual MsgX* Clone() const override																		    \
        {                                                                                                               \
            return new Msg_##tpcName(*this);                                                                            \
        };
        //argList;
#define DEF_MSG_ITEM_BASE_RESET(tpcName)                                                                                \
        virtual Msg_##tpcName* Reset() override																		    \
        {                                                                                                               \
            this->iError = 0;
            //argInit;
#define DEF_MSG_ITEM_BASE_END(tpcName)                                                                                  \
            return this;                                                                                                \
        };						                                                                                        \
    };


#define DEF_MSG_ITEM_LIKE(id, byPost, autoTell, tpcName, tpcOrigin)													    \
    struct Msg_##tpcName: public Msg_##tpcOrigin                                                                        \
    {                                                                                                                   \
        Msg_##tpcName()                                                             				                    \
        {                                                                                                               \
            const_cast<uint16_t&>(this->uMsgId) = (Msg_Base_Id + id);													\
            const_cast<bool&>(this->isPost) = byPost;																	\
            const_cast<bool&>(this->auTell) = (byPost && autoTell);														\
            Reset();																									\
        };					                                                                                            \
        virtual MsgX* Clone() const override																		    \
        {                                                                                                               \
            return new Msg_##tpcName(*this);                                                                            \
        };						                                                                                        \
    };												                                                                    \
    constexpr uint16_t Msg_##tpcName##_Id = (Msg_Base_Id + id);


#define DEF_MSG_END		}


#endif
